package com.qf.listener;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

//@EnableAsync // 开启异步处理
@Component
public class OrderQueuelistener {

    @RabbitListener(queues = "order-queue")
    public void consumerMsg(String msg, Channel channel, Message message) throws IOException {
        System.out.println("订单队列的监听者消费了数据:" + msg);
        if ("hello".equals(msg)) {
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } else if ("error".equals(msg)) {
            // 模拟这个消息消费不了
            // 如果消息本身的问题导致无法消费，我们需要记录这个消息无法消费，但是要保证消息不能丢失。
            // 消费者拒绝消费的，有不想压到队列的头部，这些消息成为死信
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
        }
    }

    // 创建一个线程池
    ExecutorService executorService = Executors.newFixedThreadPool(10);

    //    @Async // Sprin提供的异步编程的注解
    @RabbitListener(queues = "dead-queue")
    public void consumerMsgDeadQueue(String msg, Channel channel, Message message) throws IOException {

        // 每次有消息的时候就从线程池中申请一个线程去处理
        executorService.submit(new Runnable() {
            @Override
            public void run() {
                String now = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());
                System.out.println("死信队列时间" + now + "，内容" + msg);
                System.out.println("线程：" + Thread.currentThread().getId());
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                // 把无法消费的消息记录下来即可
                try {
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        });

    }

}
